package com.wudizaba.microservicesdemosystem.util;

import com.wudizaba.microservicesdemocommon.constant.ExceptionsConstant;
import com.wudizaba.microservicesdemocommon.constant.MqttConstant;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;


/**
 * @author wudizaba
 * @date 2024/06/17 16:05
 */
@Slf4j
public class MqttUtils {



    //创建客户端并建立连接
    public MqttClient createClient(String broker, String clientId, String userName, String password) {
        MqttClient mqttClient = null;
        try {
            mqttClient = new MqttClient(broker, clientId);
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            options.setUserName(userName);
            options.setPassword(password.toCharArray());
            options.setConnectionTimeout(10);
            options.setKeepAliveInterval(20);
            mqttClient.connect(options);
            return mqttClient;
        } catch (Exception e) {
//            throw new MqttException(ExceptionsConstant.MQTT_CONNECT_ERROR);
            log.info(e.getLocalizedMessage());
        }
        return null;
    }

    //创建客户端并建立连接
    public MqttClient createClientAndCallback(String broker, String clientId, String userName, String password) {
        MqttClient mqttClient = null;
        try {
            mqttClient = new MqttClient(broker, clientId);
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(false);
            options.setUserName(userName);
            options.setPassword(password.toCharArray());
            options.setConnectionTimeout(60);
            options.setKeepAliveInterval(60);
            setCallback(mqttClient);
            mqttClient.connect(options);
            return mqttClient;
        } catch (Exception e) {
//            throw new MqttException(ExceptionsConstant.MQTT_CONNECT_ERROR);
            log.info(e.getLocalizedMessage());
        }
        return null;
    }

    //订阅主题
    public  void subscribeTopic(MqttClient mqttClient, String topic, Integer qos) {
        try {
            mqttClient.subscribe(topic, qos);
            log.info(MqttConstant.SUBSCRIBE_SUCCESS);
        } catch (Exception e) {
            log.info(e.getLocalizedMessage());
        }
    }

    //发布消息
    public  void publishMessage(MqttClient mqttClient, String topic, String message, Integer qos) {
        try {
            mqttClient.publish(topic, message.getBytes(), qos, false);
            log.info(MqttConstant.PUBLISH_SUCCESS);
            disconnectClient(mqttClient);
        } catch (Exception e) {
//            throw new MqttException(ExceptionsConstant.MQTT_PUBLISH_ERROR);
            log.info(e.getLocalizedMessage());
        }
    }

    //断开连接+关闭客户端
    public  void disconnectClient(MqttClient mqttClient) {
        try {
            mqttClient.disconnect();
            mqttClient.close();
            log.info(MqttConstant.DISCONNECT_SUCCESS);
        } catch (Exception e) {
//            throw new MqttException(ExceptionsConstant.MQTT_DISCONNECT_ERROR);
            log.info(e.getLocalizedMessage());
        }
    }


    //设置回调
    public  void setCallback(MqttClient mqttClient) {
        mqttClient.setCallback(new MqttCallback() {
            //连接丢失
            @Override
            public void connectionLost(Throwable cause) {
                log.info(cause.getLocalizedMessage());
//                throw new MqttException(ExceptionsConstant.MQTT_CONNECTION_LOST);
                log.info(ExceptionsConstant.MQTT_CONNECTION_LOST);
                //重连

            }

            //接收到消息
            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                log.info(MqttConstant.RECEIVE_MESSAGE + topic);
                String messageStr = new String(message.getPayload());
                int qos = message.getQos();
                log.info(MqttConstant.RECEIVE_MESSAGE + messageStr);
                //将消息存入redis
//                redisTemplate.opsForValue().set(topic + qos, messageStr);
            }

            //消息发送成功
            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                log.info(MqttConstant.DELIVERY_SUCCESS);
            }
        });
    }

}
